package com.it.word_count;

import com.it.word_count.util.StreamUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 基于无界流处理实现的word count.
 * <p>
 * 使用netcat来实时的发送数据流.
 * command: nc -kl 7777
 *
 * @author code1997
 */
public class StreamUnboundWordCount {

    public static final String HOST_NAME = "host";
    public static final String PORT = "port";

    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.读取数据源文件
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String hostname = parameterTool.get(HOST_NAME);
        int port = parameterTool.getInt(PORT);
        DataStreamSource<String> socketTextStream = executionEnvironment.socketTextStream(hostname, port);
        StreamUtils.printWordCount(socketTextStream);
        //5.启动执行
        executionEnvironment.execute();
    }
}
